package com.wll.util.utils;

import org.jdeferred.Deferred;
import org.jdeferred.DonePipe;
import org.jdeferred.Promise;
import org.jdeferred.impl.DefaultDeferredManager;
import org.jdeferred.impl.DeferredObject;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * <pre>
 * 
 * </pre>
 *
 * @author as-chennan
 * @version AsynUtils.java v0.1  
 * @time 2017年5月22日 上午11:25:21
 * 
 * Promise<List<StockOverviewResponse>, Exception, Object> content = AsynUtils.createPromise(
 * 		new ArrayList<StockOverviewResponse>()
 *		, () -> iStockService.findAllEquipmentByCondition(daoCriterias,getMaxResult(page),getFirstResult(page)));
 *
 * Promise<Integer, Exception, Object> count = AsynUtils.createPromise(0
 *		, () -> iStockService.countEquipByCondition(daoCriterias));
 *
 * List<IResult<Object>> list = AsynUtils.runPromise(content, count);
 *
 *List<DaoCriteria> daoCriterias = buildDaoCriteriasWithOutRoles(queryRequests);
 *            CompletableFuture<List<StockOverviewResponse>> futureList = AsynUtils.submit(() ->
 *                    iStockService.findAllEquipmentByCondition(daoCriterias, getMaxResult(page), getFirstResult(page)));
 *
 *            CompletableFuture<Integer> num = AsynUtils.submit(() -> iStockService.countEquipByCondition(daoCriterias));
 *            PageImpl<StockOverviewResponse> stockOverviewResponses = null;
 *
 *            while (!(futureList.isDone() && num.isDone())) {
 *                Thread.sleep(1000);
 *            }
 *            stockOverviewResponses = new PageImpl<>(AsynUtils.get(futureList, 8, new ArrayList<>())
 *                    , page
 *                    , AsynUtils.get(num, 8, 0));
 */
public class AsynUtils {
	
	private static DefaultDeferredManager dm = new DefaultDeferredManager();

	/**
	 * 提交一个生产事件，让其变成异步处理事件
	 * 
	 * @param supplier
	 * @return
	 * 
	 * 该方法只要提交生产事件后，该生产事件就会立即让其他线程执行(参见 AsynUtils.run)，如果不需要该线程执行结果，
	 * 则不需要调用AsynUtils.get()获取结果。
	 * 
	 */
	public static <T> CompletableFuture<T> submit(Supplier<T> supplier) {
		CompletableFuture<T> future = CompletableFuture.supplyAsync(supplier, dm.getExecutorService());
		future.whenComplete((value, exception) -> {
			if (null != exception) {
				LoggUtil.error(AsynUtils.class, "执行失败！", exception);
			}
		});
		
		return future;
	}
	
	/**
	 * 等待获取异步处理结果 (结合submit使用)
	 * 
	 * @param future
	 * @param seconds 超时时间(秒)
	 * @param defaultValue 缺省值(秒)
	 * @return
	 */
	public static <T> T get(Future<T> future, int seconds, T defaultValue) {
		try {
            return future.get(seconds, TimeUnit.SECONDS);
        } catch (Exception e) {
        	LoggUtil.error(AsynUtils.class, "等待获取异步处理结果失败！", e);
            return defaultValue;
        }
	}
	
	/**
	 * 异步执行方法
	 * 
	 * @param params
	 * @param consumer
	 * 
	 * 参见 AsynUtils.submit
	 */
	public static <T> void run(T params, Consumer<T> consumer) {
		dm.getExecutorService().execute(() -> consumer.accept(params));
	}
	
	/**
	 * 异步执行方法
	 * 
	 * @param run
	 */
	public static <T> void run(Runnable run) {
		dm.getExecutorService().execute(run);
	}
	
	/**
	 * 创建异步promise
	 * 
	 * @param result
	 * @param supplier
	 * @return
	 */
	public static <T, R> Promise<T, Exception, Object> createPromise(T result, Supplier<T> supplier) {
		Deferred<T, Exception, Object> deferred = new DeferredObject<>(); 
		deferred.resolve(result);
		return deferred.promise().then(new DonePipe<T, T, Exception, Object>() {
			@Override
			public Promise<T, Exception, Object> pipeDone(T result) {
				try {
					T r = supplier.get();
					return new DeferredObject<T, Exception, Object>().resolve(r);
				} catch (Exception e) {
					return new DeferredObject<T, Exception, Object>().reject(e);
				}
			}
		}).fail(e -> LoggUtil.error(AsynUtils.class, "Promise执行失败！", e));
	}
	
	/**
	 * 处理promise(联合创建异步promise使用)
	 * 
	 * @param promises
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public static <T> List<IResult<T>> runPromise(Promise<?, ?, ?>... promises) {
		Assert.notNull(promises, "Promise不能为空");
		List<IResult<T>> list = new ArrayList<>();
		dm.when(promises).done(result -> {
			result.forEach(r -> list.add(new ResultImpl<T>((T)r.getResult())));
		});
		return list;
	}
	
	public static <R> Callable<R> createCallable(final R result) {
		return () -> result;
	}

	@SuppressWarnings("unchecked")
	public static <T> List<IResult<T>> runCallables(Callable<?>... callables) {
		Assert.notNull(callables, "Callable不能为空");
		List<IResult<T>> list = new ArrayList<>();
		
		dm.when(callables).done(result -> {
			result.forEach(r -> list.add(new ResultImpl<T>((T)r.getResult())));
		});
		
		return list;
	}
	
	@SuppressWarnings("unchecked")
	public static <T> List<IResult<T>> runCallables1(Callable<?>... callables) {
		Assert.notNull(callables, "Callable不能为空");
		ExecutorService es = dm.getExecutorService();
		List<IResult<T>> list = new ArrayList<>();
		for(int i = 0; i < callables.length; i++) {
			es.submit(callables[i]);
		}
		
		dm.when(callables).done(result -> {
			result.forEach(r -> list.add(new ResultImpl<T>((T)r.getResult())));
		});
		
		return list;
	}
	
	public static void shutdown() {
		dm.getExecutorService().shutdown();
	}
	
}
